package com.witsight.cassandra.dao.impl;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cassandra.core.ResultSetExtractor;
import org.springframework.cassandra.core.RowCallback;
import org.springframework.cassandra.core.RowCallbackHandler;
import org.springframework.context.annotation.Primary;
import org.springframework.dao.DataAccessException;
import org.springframework.data.cassandra.core.CassandraConverterRowCallback;
import org.springframework.data.cassandra.core.CassandraOperations;
import org.springframework.data.cassandra.repository.CassandraRepository;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.exceptions.DriverException;
import com.witsight.cassandra.bean.NameQuery;
import com.witsight.cassandra.dao.CassandraCommonRepository;
import com.witsight.cassandra.dao.NameQueriesRepository;
import com.witsight.cassandra.dao.PageCassandraOperations;
import com.witsight.cassandra.util.CqlBuilder;
import com.witsight.platform.model.BoEntity;
import com.witsight.platform.model.IBaseObject;
import com.witsight.platform.util.lang.BeanUtil;
import com.witsight.platform.util.lang.StringUtil;
import com.witsight.platform.util.tools.JsonUtil;

import joinery.DataFrame;

/**   
 * 说明：cassandra 不带实体的实现
 * @Title: CassandraRepositoryImpl.java 
 * @Package com.witsight.cassandra.dao.impl 
 * @See: {@link CassandraRepository}
 * Copyright: Copyright (c) 2017
 * Company:sany huax witsight team by product
 * @author: penght  
 * @date: 2017年9月25日 上午9:09:37 
 * @version: V1.0
 *
 */
@Component
@Primary
public class CassandraCommonRepositoryImpl implements CassandraCommonRepository{
	/**
	 * 说明：定义后能再同一类中使用static常量log
	 */
	private static final Logger logger = LoggerFactory.getLogger(CassandraCommonRepositoryImpl.class);
	@Autowired
	private CassandraOperations cassandraOperations;
	@Autowired
	private NameQueriesRepository nameQueries;
	
	private PageCassandraOperations getPageCassandraOperations() {
		//Assert.isTrue(!(cassandraOperations instanceof PageCassandraOperations), "cassandraOperations must be PageCassandraOperations type.");
		return (PageCassandraOperations)cassandraOperations;
	}
	
	@Override
	public <O> List<O> findSqlAllByParams(String queryName, IBaseObject params) {	
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName, bparam);
		Assert.hasText(query, "CQL must not be empty");
		NameQuery nameQuery = nameQueries.getNameQuery(queryName);
		final List<O> resluts = new ArrayList<O>();
		
		PageCassandraOperations operations = getPageCassandraOperations();
		operations.query(query, bparam, new RowCallbackHandler() {

			@SuppressWarnings("unchecked")
			@Override
			public void processRow(Row row) throws DriverException {
				if (nameQuery == null || nameQuery.getResult() == null ) {
					resluts.add((O)operations.toBoEntity(row));
				} else {
					RowCallback<O> converter = new CassandraConverterRowCallback<O>(operations.getConverter(),(Class<O>) nameQuery.getResult());
					resluts.add(converter.doWith(row));
				}
			}
		});
		
		return resluts;
	}

	@Override
	public <O> O findSqlOneByParams(String queryName, IBaseObject params) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName, bparam);
		Assert.hasText(query, "CQL must not be empty");
		NameQuery nameQuery = nameQueries.getNameQuery(queryName);
		PageCassandraOperations operations = getPageCassandraOperations();
		return operations.query(query, bparam,  new ResultSetExtractor<O>() {

			@SuppressWarnings("unchecked")
			@Override
			public O extractData(ResultSet rs) throws DriverException, DataAccessException {
				Row row = rs.one();
				if (row == null) {
					if (logger.isInfoEnabled()) {
						logger.info("ResultSet is empty. so, return null.");
					}
					return null;
				}
				if (nameQuery == null || nameQuery.getResult() == null ) {
					return (O)operations.toBoEntity(row);
				} else {
					RowCallback<O> converter = new CassandraConverterRowCallback<O>(operations.getConverter(),((Class<O>) nameQuery.getResult()));
					return converter.doWith(row);
				}	
			}

		});
	}
	
	@Override
	public <O> Page<O> findSqlPageByParams(String queryName, IBaseObject params, Pageable pageable) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName, bparam);
		Assert.hasText(query, "CQL must not be empty");
		NameQuery nameQuery = nameQueries.getNameQuery(queryName);
		PageCassandraOperations operations = getPageCassandraOperations();
		// 转化为统计cql
		String countCql = CqlBuilder.createCountQueryFor(query);
		long count = operations.selectCount(countCql);
		if (count <= 0) {
			return new PageImpl<O>(Collections.<O> emptyList());
		}
		return operations.queryPage(query, bparam, pageable, new ResultSetExtractor<Page<O>>() {

			@SuppressWarnings("unchecked")
			@Override
			public Page<O> extractData(ResultSet rs) throws DriverException, DataAccessException {
				if (rs == null) {
					if (logger.isInfoEnabled()) {
						logger.info("ResultSet is empty. so, return empty PageImpl.");
					}
					return new PageImpl<O>(Collections.<O> emptyList());
				}
				List<O> list = new ArrayList<>();
				rs.forEach(row->{
					if (nameQuery == null || nameQuery.getResult() == null ) {
						list.add((O)operations.toBoEntity(row));
					} else {
						RowCallback<O> converter = new CassandraConverterRowCallback<O>(operations.getConverter(),((Class<O>) nameQuery.getResult()));
						list.add(converter.doWith(row));
					}
				});
				
				return new PageImpl<O>(list, pageable, count);
			}
		});
		
	}
	
	@Override
	public String findPageWithJson(String queryName, IBaseObject params, Pageable pageable) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		// 转化为统计cql
		String countCql = CqlBuilder.createCountQueryFor(query);
		PageCassandraOperations operations = getPageCassandraOperations();
		long count = operations.selectCount(countCql);
		if (count <= 0) {
			return "{\"content\":[],\"totalPages\":\"0\",\"totalElements\":\"0\",\"last\":false,\"number\":\""+pageable.getPageNumber()+"\",\"size\":\""+pageable.getPageSize()+"\",\"first\":false,\"numberOfElements\":\"0\"}";
		}
		List<String> list = operations.queryPage(query, bparam, pageable, new RowCallback<String>() {
			@Override
			public String doWith(Row row) {
				RowCallback<String> converter = new CassandraConverterRowCallback<String>(operations.getConverter(), String.class);
				return converter.doWith(row);
			}
		});
		return JsonUtil.toJsonString("[".concat(String.join(",", list)).concat("]"), pageable, count);
	}

	@Override
	public String findAllWithJson(String queryName, IBaseObject params) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		final StringBuffer json = new StringBuffer("[");
		PageCassandraOperations operations = getPageCassandraOperations();
		operations.query(query, bparam, new RowCallbackHandler() {

			@Override
			public void processRow(Row row) throws DriverException {
				RowCallback<String> converter = new CassandraConverterRowCallback<String>(operations.getConverter(), String.class);
				json.append(converter.doWith(row)).append(",");
			}
			
		});
		return StringUtil.removeEnd(json.toString(), ",") + "]";
	}

	@Override
	public DataFrame<Object> findDataFramesByParams(String queryName, IBaseObject params) {
		List<BoEntity> list = findSqlAllByParams(queryName,params);
		Set<String> columns = new HashSet<String>();
		if(list!=null&&list.size()>0)
			columns = list.get(0).keySet();
		DataFrame<Object> dataFrame = new DataFrame<Object>(columns);
		for(BoEntity obj:list) {
			List<Object> rowData = new ArrayList<>();
			columns.forEach(column->{
				rowData.add((Object) obj.get(column));
			});
			dataFrame.append(rowData);
		}
		return dataFrame;
	}
}
